package com.eric.demo.kafka;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonIgnore;
import org.apache.commons.lang3.StringUtils;

import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;

/**
 * 
* @Description: kafka接受到的消息
* @author wugaoqiang
* @date 2018年6月15日 上午10:40:24 
* @version
 */
public class Wares implements Cloneable {
    /**
     * 数据库的schema
     */
    private String schema;

    /**
     * 数据库的table
     */
    private String table;
    /**
     * 数据库的操作 INSERT UPDATE DELETE
     */
    private String operation;

    /**
     * 表中该行记录在operation之前的数据
     */
    private HashMap<String, String> beforeColumn;

    /**
     * 表中该行记录在operation之后的数据
     */
    private HashMap<String, String> afterColumn;

    /**
     * binlog里记录变更发生的时间，精确到秒
     */
    private Date binlogExecuteTime;

    /**
     * 生产者发送消息到kafka的时间
     */
    private Date sendTime;
    /**
     * 消费者从kafka获取到消息的时间
     */
    private Date receiveTime;
    /**
     * 重试次数，0表示发送一次就成功，未发生重试
     */
    private Integer retryCount;
    /**
     * 延迟时间，binlogExecuteTime和消费者获取到消息的时间间隔（receiveTime - binlogExecuteTime）
     */
    private Long delayTime;

    @Deprecated
    @JsonIgnore
    private Long binlogPostiton;

    @Deprecated
    @JsonIgnore
    private String binlogFileName;

    @Deprecated
    @JsonIgnore
    private String topic;
    
    @Override
	public String toString() {
		return "Wares [schema=" + schema + ", table=" + table + ", operation=" + operation + ", beforeColumn="
				+ beforeColumn + ", afterColumn=" + afterColumn + ", binlogExecuteTime=" + binlogExecuteTime
				+ ", sendTime=" + sendTime + ", receiveTime=" + receiveTime + ", retryCount=" + retryCount
				+ ", delayTime=" + delayTime + ", binlogPostiton=" + binlogPostiton + ", binlogFileName="
				+ binlogFileName + ", topic=" + topic + "]";
	}

	public String getSchema() {
        return schema;
    }

    public void setSchema(String schema) {
        this.schema = schema;
    }

    public String getTable() {
        return table;
    }

    public void setTable(String table) {
        this.table = table;
    }

    public String getOperation() {
        return operation;
    }

    public void setOperation(String operation) {
        this.operation = operation;
    }

    public HashMap<String, String> getBeforeColumn() {
        return beforeColumn;
    }

    public void setBeforeColumn(HashMap<String, String> beforeColumn) {
        this.beforeColumn = beforeColumn;
    }

    public HashMap<String, String> getAfterColumn() {
        return afterColumn;
    }

    public void setAfterColumn(HashMap<String, String> afterColumn) {
        this.afterColumn = afterColumn;
    }

    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    public Date getBinlogExecuteTime() {
        return binlogExecuteTime;
    }

    public void setBinlogExecuteTimeByLong(long binlogExecuteTime) {
        this.binlogExecuteTime = new Date(binlogExecuteTime);
    }

    public void setBinlogExecuteTime(Date binlogExecuteTime) {
        this.binlogExecuteTime = binlogExecuteTime;
    }

    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    public Date getSendTime() {
        return sendTime;
    }

    public void setSendTime(Date sendTime) {
        this.sendTime = sendTime;
    }


    public void setReceiveTime(Date receiveTime) {
        this.receiveTime = receiveTime;
    }

    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    public Date getReceiveTime() {
        return receiveTime;
    }

    public Integer getRetryCount() {
        return retryCount;
    }

    public void setRetryCount(Integer retryCount) {
        this.retryCount = retryCount;
    }

    public Integer getIntegerValueFromBeforeColumn(String columnName) {
        if (getBeforeColumn() != null && getBeforeColumn().containsKey(columnName)) {
            if (StringUtils.isNotBlank(getBeforeColumn().get(columnName))) {
                return Integer.valueOf(getBeforeColumn().get(columnName));
            }
        }
        return null;
    }

    public Integer getIntegerValueFromAfterColumn(String columnName) {
        if (getAfterColumn() != null && getAfterColumn().containsKey(columnName)) {
            if (StringUtils.isNotBlank(getAfterColumn().get(columnName))) {
                return Integer.valueOf(getAfterColumn().get(columnName));
            }
        }
        return null;
    }

    public Long getLongValueFromBeforeColumn(String columnName) {
        if (getBeforeColumn() != null && getBeforeColumn().containsKey(columnName)) {
            if (StringUtils.isNotBlank(getBeforeColumn().get(columnName))) {
                return Long.valueOf(getBeforeColumn().get(columnName));
            }
        }
        return null;
    }

    public Long getLongValueFromAfterColumn(String columnName) {
        if (getAfterColumn() != null && getAfterColumn().containsKey(columnName)) {
            if (StringUtils.isNotBlank(getAfterColumn().get(columnName))) {
                return Long.valueOf(getAfterColumn().get(columnName));
            }
        }
        return null;
    }

    public String getStringValueFromBeforeColumn(String columnName) {
        if (getBeforeColumn() != null && getBeforeColumn().containsKey(columnName)) {
            if (getBeforeColumn().get(columnName) != null) {
                return getBeforeColumn().get(columnName);
            }
        }
        return "";
    }

    public String getStringValueFromAfterColumn(String columnName) {
        if (getAfterColumn() != null && getAfterColumn().containsKey(columnName)) {
            if (getAfterColumn().get(columnName) != null) {
                return getAfterColumn().get(columnName);
            }
        }
        return "";
    }

    public Boolean getBooleanValueFromBeforeColumn(String columnName) {
        if (getBeforeColumn() != null && getBeforeColumn().containsKey(columnName)) {
            if (StringUtils.isNotBlank(getBeforeColumn().get(columnName))) {
                return Boolean.valueOf(getBeforeColumn().get(columnName));
            }
        }
        return null;
    }

    public Boolean getBooleanValueFromAfterColumn(String columnName) {
        if (getAfterColumn() != null && getAfterColumn().containsKey(columnName)) {
            if (StringUtils.isNotBlank(getAfterColumn().get(columnName))) {
                return Boolean.valueOf(getAfterColumn().get(columnName));
            }
        }
        return null;
    }

    public Date getDateValueFromBeforeColumn(String columnName) {
        if (getBeforeColumn() != null && getBeforeColumn().containsKey(columnName)) {
            if(StringUtils.isNotBlank(getBeforeColumn().get(columnName))){
                SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                Date date = null;
                try {
                    date = formatter.parse(getBeforeColumn().get(columnName));
                } catch (ParseException e) {
                }
                return date;
            }

        }
        return null;
    }

    public Date getDateValueFromAfterColumn(String columnName) {
        if (getAfterColumn() != null && getAfterColumn().containsKey(columnName)) {
            if(StringUtils.isNotBlank(getAfterColumn().get(columnName))){
                SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                Date date = null;
                try {
                    date = formatter.parse(getAfterColumn().get(columnName));
                } catch (ParseException e) {
                    e.printStackTrace();
                }
                return date;
            }
        }
        return null;
    }

    public BigDecimal getBigDecimalValueFromBeforeColumn(String columnName) {
        if (getBeforeColumn() != null && getBeforeColumn().containsKey(columnName)) {
            if (StringUtils.isNotBlank(getBeforeColumn().get(columnName))) {
                return new BigDecimal(getBeforeColumn().get(columnName));
            }
        }
        return null;
    }

    public BigDecimal getBigDecimalValueFromAfterColumn(String columnName) {
        if (getAfterColumn() != null && getAfterColumn().containsKey(columnName)) {
            if (StringUtils.isNotBlank(getAfterColumn().get(columnName))) {
                return new BigDecimal(getAfterColumn().get(columnName));
            }
        }
        return null;
    }

    public void calculateDelayTimeAndSet(Date binlogExecuteTime, Date receiveTime) {
        this.delayTime = receiveTime.getTime() - binlogExecuteTime.getTime();
    }

    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    public Long getDelayTime() {
        return delayTime;
    }

    public void setDelayTime(Long delayTime) {
        this.delayTime = delayTime;
    }

    public Long getBinlogPostiton() {
        return binlogPostiton;
    }

    public void setBinlogPostiton(Long binlogPostiton) {
        this.binlogPostiton = binlogPostiton;
    }

    public String getBinlogFileName() {
        return binlogFileName;
    }

    public void setBinlogFileName(String binlogFileName) {
        this.binlogFileName = binlogFileName;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    @Override
    public Object clone(){
        try {
            return super.clone();
        } catch (CloneNotSupportedException e) {
            throw new RuntimeException(e);
        }
    }
    
    public enum OperationType {
    	/**
    	 * 插入
    	 */
        INSERT("INSERT"),
        /**
    	 * 更新
    	 */
        UPDATE("UPDATE"),
        /**
    	 * 删除
    	 */
        DELETE("DELETE");

        private String type;

        OperationType(String type) {
            this.type = type;
        }

        public String desc() {
            return type;
        }
    }
}
